package oz.mano.online.zookeeper;

import java.io.UnsupportedEncodingException;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.CuratorFrameworkFactory.Builder;
import org.apache.curator.framework.api.ACLProvider;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;

public class ZKClientUtils {
	
	private CuratorFramework client;
	
	private ZKConfig zkConfig;
	
	public void init() throws UnsupportedEncodingException{
		if(null != client){
			return ;
		}
		if (client != null) return;
        if (zkConfig == null) {
            zkConfig = ZKConfig.build();
        }
        Builder builder = CuratorFrameworkFactory
                .builder()
                .connectString(zkConfig.getHosts())
                .retryPolicy(new ExponentialBackoffRetry(zkConfig.getBaseSleepTimeMs(), zkConfig.getMaxRetries(), zkConfig.getMaxSleepMs()))
                .namespace(zkConfig.getNamespace());

        if (zkConfig.getConnectionTimeout() > 0) {
            builder.connectionTimeoutMs(zkConfig.getConnectionTimeout());
        }
        if (zkConfig.getSessionTimeout() > 0) {
            builder.sessionTimeoutMs(zkConfig.getSessionTimeout());
        }

        if (zkConfig.getDigest() != null) {
            /*
             * scheme对应于采用哪种方案来进行权限管理，zookeeper实现了一个pluggable的ACL方案，可以通过扩展scheme，来扩展ACL的机制。
             * zookeeper缺省支持下面几种scheme:
             *
             * world: 默认方式，相当于全世界都能访问; 它下面只有一个id, 叫anyone, world:anyone代表任何人，zookeeper中对所有人有权限的结点就是属于world:anyone的
             * auth: 代表已经认证通过的用户(cli中可以通过addauth digest user:pwd 来添加当前上下文中的授权用户); 它不需要id, 只要是通过authentication的user都有权限（zookeeper支持通过kerberos来进行authencation, 也支持username/password形式的authentication)
             * digest: 即用户名:密码这种方式认证，这也是业务系统中最常用的;它对应的id为username:BASE64(SHA1(password))，它需要先通过username:password形式的authentication
             * ip: 使用Ip地址认证;它对应的id为客户机的IP地址，设置的时候可以设置一个ip段，比如ip:192.168.1.0/16, 表示匹配前16个bit的IP段
             * super: 在这种scheme情况下，对应的id拥有超级权限，可以做任何事情(cdrwa)
             */
            builder.authorization("digest", zkConfig.getDigest().getBytes("UTF-8"));
            builder.aclProvider(new ACLProvider() {
                @Override
                public List<ACL> getDefaultAcl() {
                    return ZooDefs.Ids.CREATOR_ALL_ACL;
                }

                @Override
                public List<ACL> getAclForPath(final String path) {
                    return ZooDefs.Ids.CREATOR_ALL_ACL;
                }
            });
        }
        client = builder.build();
	}
	
	public String getData(final String key) throws UnsupportedEncodingException, Exception{
		if(isExisted(key)){
			return new String(client.getData().forPath(key),"UTF-8");
		}
		return null;
	}
	
	public boolean isExisted(final String key) throws Exception{
		return null != client.checkExists().forPath(key);
	}
	
	public List<String> getChildrenKeys(final String key) throws Exception{
		if(!isExisted(key)){
			return Collections.emptyList();
		}
		List<String> children = client.getChildren().forPath(key);
		children.sort(Comparator.reverseOrder());
		return children;
	}
	
	public void removeData(final String key) throws Exception{
		client.delete().deletingChildrenIfNeeded().forPath(key);
	}
	
	public void watch(final String key) throws Exception{
		ExecutorService tp = Executors.newFixedThreadPool(2);
		System.out.println(Thread.currentThread().getName());
		PathChildrenCache cache = new PathChildrenCache(client,key, true, false, tp);
		cache.start(PathChildrenCache.StartMode.NORMAL);
		cache.getListenable().addListener(new PathChildrenCacheListener() {
			
			@Override
			public void childEvent(CuratorFramework client, PathChildrenCacheEvent event)
					throws Exception {
				switch(event.getType()){
				case CHILD_ADDED:
					// 添加 节点 ，处理业务逻辑
				    break;
				case CHILD_UPDATED:
					// 更新 节点 ，处理业务逻辑
				    break;
				case CHILD_REMOVED:
					//删除 节点 ，处理业务逻辑
				    break;    
				default:
				    //...;
				    break;
				}
				
			}
		});
	}
	
	 /**
     * 持久化数据
     *
     * @param key
     * @param value
	 * @throws Exception 
     */
    public void registerPersist(final String key, final String value) throws Exception {
            if (isExisted(key)) {
                update(key, value);
            } else {
                client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(key, value.getBytes());
            }
    }

    /**
     * 更新数据
     *
     * @param key
     * @param value
     * @throws Exception 
     * @throws UnsupportedEncodingException 
     */
    public void update(final String key, final String value) throws UnsupportedEncodingException, Exception {
            /*TransactionOp op = client.transactionOp();
            client.transaction().forOperations(
                    op.check().forPath(key),
                    op.setData().forPath(key, value.getBytes(Constants.UTF_8))
            );*/
            client.inTransaction().check().forPath(key).and().setData().forPath(key, value.getBytes("Constants.UTF_8")).and().commit();
    }

    /**
     * 注册临时数据
     *
     * @param key
     * @param value
     * @throws Exception 
     */
    public void registerEphemeral(final String key, final String value, boolean cacheNode) throws Exception {
            if (isExisted(key)) {
                client.delete().deletingChildrenIfNeeded().forPath(key);
            }
            client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(key, value.getBytes("UTF-8"));
        
    }

    /**
     * 注册临时数据
     *
     * @param key
     * @param value
     * @throws Exception 
     */
    public void reRegisterEphemeral(final String key, final String value) throws Exception {
        registerEphemeral(key, value, false);
    }

    /**
     * 注册临时数据
     *
     * @param key
     * @param value
     * @throws Exception 
     */
    public void registerEphemeral(final String key, final String value) throws Exception {
        registerEphemeral(key, value, true);
    }

    /**
     * 注册临时顺序数据
     *
     * @param key
     * @param value
     * @param cacheNode 第一次注册时设置为true, 连接断开重新注册时设置为false
     * @throws Exception 
     */
    private void registerEphemeralSequential(final String key, final String value, boolean cacheNode) throws Exception {
        client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(key, value.getBytes());
    }

    private void reRegisterEphemeralSequential(final String key, final String value) throws Exception {
        registerEphemeralSequential(key, value, false);
    }

    public void registerEphemeralSequential(final String key, final String value) throws Exception {
        registerEphemeralSequential(key, value, true);
    }


    /**
     * 注册临时顺序数据
     *
     * @param key
     * @throws Exception 
     */
    public void registerEphemeralSequential(final String key) throws Exception {
            client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(key);
    }
	

}
